package com.zst.kafka.spring;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.MessageListener;

import java.util.HashMap;
import java.util.Map;

@SpringBootApplication
public class Main {
    public static void main(String[] args) {
        org.springframework.boot.SpringApplication.run(Main.class, args);

        Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put("bootstrap.servers", "192.168.123.46:19003");
        DefaultKafkaConsumerFactory<String, String> defaultKafkaConsumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);

        ContainerProperties properties = new ContainerProperties("test-topic");
        properties.setMessageListener(new MessageListener<String, String>() {
            @Override
            public void onMessage(ConsumerRecord<String, String> data) {
                System.out.println(data.value());
            }
        });
        properties.setGroupId("1");

        ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(defaultKafkaConsumerFactory, properties);
        container.start();
    }


//    @Bean
//    public ApplicationRunner producer(KafkaTemplate<String, String> kafkaTemplate
//        return args -> {
//            kafkaTemplate.send("test-topic", "hello1");
//            kafkaTemplate.send("test-topic", "hello2");
//        };
//    }

}
